package com.lovecaa.broadcast;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.nio.charset.StandardCharsets;

/**
 * 消费者
 * 创建时间： 2023年10月20日 23:32:00
 *
 * @author Mr.Tao
 * @since 1.0.0
 */
public class BroadcastConsumer2 {
    public static void main(String[] args) throws MQClientException {
        // 广播消费者组2
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("BroadcastConsumerGroup2");
        consumer.setNamesrvAddr("192.168.192.110:9876");
        // 设置消费模式为广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 广播消息多个消费者组订阅同一个主题
        consumer.subscribe("Test1", "*");
        // 这里采用的是一个并发消费的监听器
        consumer.setMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            for (int i = 0; i < list.size(); i++) {
                System.out.println(i + "消息消费成功:" + new String(list.get(i).getBody(), StandardCharsets.UTF_8));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.out.println("消费者启动成功...");
    }
}
